package com.qkjia.consumer;


import com.qkjia.constant.KafkaTopicConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
@Slf4j
@ConditionalOnExpression("'${spring.profiles.active}'.equals('test')")
public class LocalizationConsumer {

//    @KafkaListener(topics = KafkaTopicConstant.TEST_TOPIC_01,groupId = "testGroup01")
    @KafkaListener(topics = KafkaTopicConstant.TEST_TOPIC_01)
    public void deviceTest(ConsumerRecord consumerRecord, Acknowledgment ack) {
        Object value = consumerRecord.value();
        log.info("deviceValue="+value);
        ack.acknowledge();
    }

    @KafkaListener(topics = KafkaTopicConstant.SOURCE_TOPIC_02)
    public void deviceSource(ConsumerRecord consumerRecord, Acknowledgment ack) {
        Object value = consumerRecord.value();
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info("deviceValue="+value);
        ack.acknowledge();
    }
//
//    @KafkaListener(topics = "testTopic03",groupId = "testGroup03")
//    public void topic03(ConsumerRecord consumerRecord, Acknowledgment ack) {
//        Object value = consumerRecord.value();
//        try {
//            Thread.sleep(10);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
//        log.info("deviceValue="+value);
//        ack.acknowledge();
//    }
}
